package com.burges.net.dataSet.api.transform

import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment}
import org.apache.flink.util.Collector

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/2/11 
  * 描述      多表关联JOIN操作
  */
object JoinOperationDemo {

	def main(args: Array[String]): Unit = {
		var environment: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
		// 创建两个数据集
		val tupleDs1: DataSet[(Int, String)] = environment.fromElements((1,"a"),(1,"a"),(1,"a"))
		val tupleDs2: DataSet[(Double, Int)] = environment.fromElements((1.1,1),(1.1,1),(1.1,1))
		// JOIN TUPLE
		val tupleRs = tupleDs1.join(tupleDs2).where(0).equalTo(1)

		// 两个CaseClass
		val caseClassPersonDs1: DataSet[Person] = environment.fromElements(Person(1, "a"),Person(1, "a"))
		val caseClassPersonDs2: DataSet[(Double, Int)] = environment.fromElements((12.3,1), (22.3,3))
		// 第一种使用方式
		val caseClassRs = caseClassPersonDs1.join(caseClassPersonDs2).where("age").equalTo(1)

		// 第二种使用方式
		var caseClassRs2 = caseClassPersonDs1.join(caseClassPersonDs2).where("age").equalTo(1){
			(left, right) => (left.age, left.name, right._1 +1)
		}

		// 第三种使用方式
		caseClassPersonDs1.join(caseClassPersonDs2).where("age").equalTo(1){
			(left, right, collector: Collector[(String, Double)]) =>
				collector.collect(left.name, right._1+1)
				collector.collect("prefix_" + left.name, right._1+1)
		}

		/**
		  * 通过SizeHint标记数据集的大小
		  */
		// 提示Flink第二个数据集是小数据集
		caseClassPersonDs1.joinWithTiny(caseClassPersonDs2).where("id").equalTo(1)
		// 提示Flink第二个数据集是大数据集
		caseClassPersonDs1.joinWithHuge(caseClassPersonDs2).where("id").equalTo(1)

		//除了上面两种方法外，还有更加灵活高效的方式
		// 将第一个数据集广播出去，转换成HashTable存储，适用于第一个数据集非常小的情况
//		caseClassPersonDs1.join(caseClassPersonDs2, BROADCAST_HASH_FIRST).where("id").equalTo(1)
		// 将第二个数据集广播出去，转换成HashTable存储，适用于第二个数据集非常小的情况
//		caseClassPersonDs1.join(caseClassPersonDs2, BROADCAST_HASH_SECOND).where("id").equalTo(1)
		// 和不设定Hint相同，将优化的工作交给系统处理
//		caseClassPersonDs1.join(caseClassPersonDs2, OPTIMIZER_CHOOSES).where("id").equalTo(1)
		// 将两个数据集重新分区，并将第一个数据集转换成HashTable存储，该策略适用于第一个数据集比第二个数据集小，但两个数据集相对比较大的情况
//		caseClassPersonDs1.join(caseClassPersonDs2, REPARTITION_HASH_FIRST).where("id").equalTo(1)
		// 将两个数据集重新分区，并将第二个数据集转换成HashTable存储，该策略适用于第二个数据集比第一个数据集小，但两个数据集相对比较大的情况
//		caseClassPersonDs1.join(caseClassPersonDs2, REPARTITION_HASH_SECOND).where("id").equalTo(1)
		// 将两个数据集重新分区，并将每个分区重新排序，该策略适用于两个数据集已经拍好顺序的情况
//		caseClassPersonDs1.join(caseClassPersonDs2, REPARTITION_SORT_MERGE).where("id").equalTo(1)
	}

}

case class Person (age: Int, name: String)
